package org.witsoft.localmq.impl;

import org.witsoft.localmq.ConfigManager;
import org.witsoft.localmq.FileWriter;
import org.witsoft.localmq.MessageBroker;
import org.witsoft.localmq.model.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @ClassName: QueuedMqMessageBroker
 * @Description:
 * @Author: Colin
 * @Date: 2020/12/26 6:25 PM
 * @Version 1.0
 **/
@Slf4j
public class QueuedMessageBroker implements MessageBroker {

    private FileWriter fileWriter;

    private Queue<Message> messageQueue;

    private AtomicLong dataSizeCounter = new AtomicLong();

    private ExecutorService threadPoolExecutor;

    private ConfigManager configManager;

    public QueuedMessageBroker(ConfigManager configManager, FileWriter fileWriter){
        this.fileWriter = fileWriter;
        this.configManager = configManager;
        int brokerThreads = configManager.getBrokerThreads();
        int brokerQueueSize = configManager.getBrokerQueueSize();
        if(brokerQueueSize > 0) {
            messageQueue = new LinkedBlockingDeque<>(brokerQueueSize);
            threadPoolExecutor = Executors.newFixedThreadPool(brokerThreads);
            for (int i = 0; i < brokerThreads; i++) {
                threadPoolExecutor.execute(new FlushMessageRunnable());
            }
        }
    }

    @Override
    public void produce(List<Message> messages) {
        int maxDataSize = configManager.getBrokerQueueMaxDataSize();
        boolean isOverSize = false;
        if(maxDataSize > 0){
            isOverSize = dataSizeCounter.get() > maxDataSize;
        }
        if(isOverSize){
            log.warn("超出队列数据最大限制[{}]，将会实时落盘!", maxDataSize);
        }
        List<Message> rejectList ;
        if(messageQueue == null || isOverSize){
            rejectList = messages;
        }else{
            rejectList = new ArrayList<>();
            for (Message message : messages) {
                if(!messageQueue.offer(message)){
                    rejectList.add(message);
                }else{
                    // 增加队列的字节大小计数
                    long dataLen = message.getData().length;
                    dataSizeCounter.accumulateAndGet(dataLen, (x,y)->x+y);
                }
            }
        }
        if(!rejectList.isEmpty()) {
            try {
                fileWriter.write(rejectList);
            } catch (IOException e) {
                log.error("消息落盘出现异常", e);
            }
        }
    }

    @Override
    public int getQueueSize(){
        return messageQueue.size();
    }

    @Override
    public long getQueueDataSize() {
        return dataSizeCounter.longValue();
    }

    public class FlushMessageRunnable implements Runnable{

        @Override
        public void run() {
            while(!Thread.interrupted()){
                List<Message> messageList = new ArrayList<>();
                for (int i = 0; i < configManager.getBrokerBatchSize() ; i++) {
                    Message message = messageQueue.poll();
                    if(message == null){
                        break;
                    }
                    messageList.add(message);
                    // 减少队列的字节大小计数
                    long dataLen = message.getData().length;
                    dataSizeCounter.accumulateAndGet(dataLen, (x,y)->x-y);
                }
                if(messageList.isEmpty()){
                    try {
                        Thread.sleep(100);
                        continue;
                    }catch (InterruptedException e){
                        break;
                    }
                }
                log.debug("消息落盘 - {}条数据" , messageList.size());
                try {
                    fileWriter.write(messageList);
                } catch (IOException e) {
                   log.error("消息落盘出现异常" ,e);
                }
            }
        }
    }

    public static void main(String[] args) {
        AtomicLong atomicLong = new AtomicLong(Long.MAX_VALUE);
        for (int i = 0; i < 10; i++) {
           long value = atomicLong.accumulateAndGet(10, (x,y)->x + y);
            System.out.println(value);
        }
    }
}
